package com.offcn.bigdata.datastream.source

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}

/**
 * @Auther: BigData-LGW
 * @ClassName: UserDefineParallelSource
 * @Date: 2020/12/15 21:56
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
object UserDefineParallelSource {
    def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val ret = env.addSource(new MySourceFunction()).setParallelism(2)

    ret.print()

    env.execute(s"${UserDefineParallelSource.getClass.getSimpleName}")
}
    class MySourceFunction extends ParallelSourceFunction[Int] {

        override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
            var num = 1
            while(true) {
                println("上游sourceFunction生产的数据：" + num)
                ctx.collect(num)
                num += 1
                Thread.sleep(500)
            }
        }

        override def cancel(): Unit = {

        }
    }
}
